package nio.reactor;


import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.IOException;
import java.net.InetSocketAddress;
import java.nio.channels.SelectionKey;
import java.nio.channels.Selector;
import java.nio.channels.ServerSocketChannel;
import java.nio.channels.SocketChannel;
import java.util.Iterator;
import java.util.Set;
import java.util.concurrent.atomic.AtomicInteger;

/**
 * @Desc:   多线程Reactor反应器的实践案例
 * @author: cww
 * @DateTime: 2020-02-08 17:18
 */

public class MultiThreadEchoServerReactor {
    private static final Logger logger = LoggerFactory.getLogger(MultiThreadEchoServerReactor.class);
    ServerSocketChannel serverSocketChannel;
    AtomicInteger next = new AtomicInteger(0);
    // 选择器集合，引入多个选择器
    Selector[] selectors = new Selector[1];
    // 引入多个子反应器
    SubReactor[] subReactors = null;

    /**
     * 初始化反应器
     * @throws IOException
     */
    public MultiThreadEchoServerReactor() throws IOException {
        // 初始化多个选择器
        selectors[0] = Selector.open();
//        selectors[1] = Selector.open();
        // 两个反应器共同监听的通道
        serverSocketChannel = ServerSocketChannel.open();
        serverSocketChannel.configureBlocking(false);
        InetSocketAddress address = new InetSocketAddress("127.0.0.1", 8000);
        serverSocketChannel.socket().bind(address);

        // 第一个选择器，负责监控新的连接事件,这里先把通道注册到 selector[0] 中去，每次轮询分别
        // 使用不同的 selector，轮询在 Handler 时将 使用不同的 selector
        SelectionKey sk = serverSocketChannel.register(selectors[0], SelectionKey.OP_ACCEPT);
        // 绑定 Handler：attach 新的 <连接监控 handler 处理器> 到 SelectionKey （选择键）
        // 第一次绑定的是 AcceptHandler，意在初始化与客户端的连接，事件是 accept ，当连接上之后，
        // 会在 AcceptHandler 的构造方法之内将 MultiThreadEchoHandler attach 进 SelectionKey 中，
        // 之后与客户端的交互就由 MultiThreadEchoHandler 来处理了
        sk.attach(new AcceptorHandler());
        // 第一个子反应器，一子反应器负责一个选择器
        SubReactor subReactor1 = new SubReactor(selectors[0]);
        // 第二个子反应器，一子反应器负责一个选择器
//        SubReactor subReactor2 = new SubReactor(selectors[1]);
//        subReactors = new SubReactor[]{subReactor1, subReactor2};
        subReactors = new SubReactor[]{subReactor1};
        logger.info("currentThread: " + Thread.currentThread() + " ,多线程Reactor反应器 启动成功----");
    }

    private void startService() {
        // 一个子反应器对应一个线程
        new Thread(subReactors[0]).start();
//        new Thread(subReactors[1]).start();
    }

    /**
     * 子反应器，子反应器负责将当前注册的通道事件 select 出来，然后再分发给 Handler
     */
    class SubReactor implements Runnable {
        // 每个线程负责一个选择器的查询和选择
        final Selector selector;

        public SubReactor(Selector selector) {
            this.selector = selector;
        }

        @Override
        public void run() {
            try {
                while (!Thread.interrupted()) {
                    // 阻塞当前线程
                    selector.select();
                    logger.info("currentThread: " + Thread.currentThread() + " ,阻塞线程 查询反应器中 IO 事件");
                    // 从当前反应器中查出所有 IO 事件
                    Set<SelectionKey> keySet = selector.selectedKeys();
                    Iterator<SelectionKey> it = keySet.iterator();
                    while (it.hasNext()) {
                        // 反应器负责 dispatch 收到的事件
                        SelectionKey sk = it.next();
                        logger.info("currentThread: " + Thread.currentThread() + " ,反应器 dispatch IO 事件");
                        dispatcher(sk);
                    }
                    // 选择键遍历完后清空
                    keySet.clear();
                }
            } catch (IOException e) {
                e.printStackTrace();
            }
        }

        void dispatcher(SelectionKey sk) {
            Runnable handler = (Runnable) sk.attachment();
            logger.info("currentThread: " + Thread.currentThread() + " ,读取并执行 SelectionKey 中的 Handler----");
            // 调用之前 attach 绑定到选择键的 handler 处理器对象
            if (handler != null) {
                handler.run();
            }
        }
    }

    /**
     * Handler 新进入连接处理器
     */
    class AcceptorHandler implements Runnable {

        @Override
        public void run() {
            SocketChannel channel = null;
            try {
                logger.info("currentThread: " + Thread.currentThread() + " ,Handler 新进入连接处理器----");
                channel =  serverSocketChannel.accept();
                if (channel != null) {
                    new MultiThreadEchoHandler(selectors[next.get()], channel);
                }
            } catch (Exception e) {
                e.printStackTrace();
                if(channel != null){
                    try {
                        channel.close();
                    } catch (IOException e1) {
                        logger.info("----通道关闭失败！！！----");
                    }
                }
            }
            if (next.incrementAndGet() == selectors.length) {
                next.set(0);
            }
        }
    }

    public static void main(String[] args) throws Exception {
        MultiThreadEchoServerReactor server = new MultiThreadEchoServerReactor();
        server.startService();
    }
}
